package com.kalvan.core.mq.consumer;

import com.kalvan.core.thread.BaseAsynchronousCmd;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

/**
 * 异步通知消费者
 *
 * @author chenliang
 */
@Slf4j
public class TransNotifyDsConsumer extends BaseAsynchronousCmd implements MessageListenerConcurrently {

    public TransNotifyDsConsumer(int corePoolSize, int maximumPoolSize, long keepAliveTime, int workQueueSize) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, workQueueSize);
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        int index = 0;
        try {
            for (; index < msgs.size(); index++) {
                MessageExt msg = msgs.get(index);
                String msgInfo = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                log.info("MQ：消费者{} 接收新信息: {} {} {} {} {}", this.getClass(), msg.getMsgId(), msg.getTopic(),
                        msg.getTags(), msg.getKeys(), msgInfo);
                while (true) {
                    int queueSize = threadPool.getQueue().size();
                    int active = threadPool.getActiveCount();
                    int poolRunCount = queueSize + active;
                    if (poolRunCount >= 650) {
                        log.warn("线程超出不执行，线程队列数： " + queueSize + "；当前活动线程数：" + active);
                        // 线程满，休眠2秒
                        Thread.sleep(2000);
                    } else {
                        break;
                    }
                }
                TransNotifyDsCmd cmd = new TransNotifyDsCmd(msgInfo);
                threadPool.execute(cmd);
            }
        } catch (Exception e) {
            log.error("消费数据异常：" + e.getMessage(), e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        } finally {
            if (index < msgs.size()) {
                context.setAckIndex(index + 1);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
